package com.visec.hdy.hadoop.fsb;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.function.Predicate;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.log4j.Logger;

import com.visec.hdy.hadoop.proxy.FileStatusProxy;
import com.visec.hdy.hadoop.standard.FSB;

/**
 * 依赖hdfs实现存储实现类
 * 
 * @author hdy
 *
 */
public class HdfsFSB implements FSB {
	private FileSystem hdfs;
	Logger l = Logger.getLogger(HdfsFSB.class);

	public HdfsFSB(String url) {
		Configuration config = new Configuration();
		config.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
		config.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
		try {
			this.hdfs = FileSystem.get(URI.create(url), config);
		} catch (IOException e) {
			l.error("", e);
		}
	}

	@Override
	public void create(File file, InputStream inputStream) {
		try {
			FSDataOutputStream create = this.hdfs.create(new Path(getPath(file)));
			IOUtils.copyBytes(inputStream, create, 2048, true);
		} catch (IllegalArgumentException | IOException e) {
			l.error("", e);
		}
	}

	@Override
	public OutputStream create(File file) {
		try {
			return this.hdfs.create(new Path(getPath(file)));
		} catch (IllegalArgumentException | IOException e) {
			l.error("", e);
		}
		return null;
	}

	@Override
	public void append(File file, InputStream inputStream) {
		try {
			FSDataOutputStream create = this.hdfs.append(new Path(getPath(file)));
			IOUtils.copyBytes(inputStream, create, 2048, true);
		} catch (IllegalArgumentException | IOException e) {
			l.error("", e);
		}
	}

	@Override
	public OutputStream append(File file) {
		try {
			return this.hdfs.append(new Path(getPath(file)));
		} catch (IllegalArgumentException | IOException e) {
			l.error("", e);
		}
		return null;
	}

	@Override
	public boolean delete(File file, boolean isRecursion) {
		try {
			return this.hdfs.delete(new Path(getPath(file)), isRecursion);
		} catch (IllegalArgumentException | IOException e) {
			l.error("", e);
		}
		return false;
	}

	@Override
	public InputStream get(File file) {
		try {
			return this.hdfs.open(new Path(getPath(file)));
		} catch (IllegalArgumentException | IOException e) {
			l.error("", e);
		}
		return null;
	}

	@Override
	public boolean isExist(File file) {
		try {
			return this.hdfs.isFile(new Path(getPath(file)));
		} catch (IllegalArgumentException | IOException e) {
			l.error("", e);
		}
		return false;
	}

	@Override
	public List<File> listFiles(File file) {
		return list(FileStatus::isFile, file);
	}

	protected List<File> list(Predicate<FileStatus> predicate, File file) {
		try {
			Path path = new Path(getPath(file));
			if (!hdfs.exists(path))
				hdfs.mkdirs(path);
			FileStatus[] listStatus = this.hdfs.listStatus(path);
			List<File> files = new ArrayList<>();
			for (FileStatus fs : listStatus) {
				if (predicate.test(fs)) {
					files.add(FileStatusProxy.create(fs));
				}
			}
			return files;
		} catch (IllegalArgumentException | IOException e) {
			l.error("", e);
		}
		return Collections.emptyList();
	}

	@Override
	public List<File> listDirs(File file) {
		return list(FileStatus::isDirectory, file);
	}

	@Override
	public void close() {
		try {
			this.hdfs.close();
		} catch (IOException e) {
			l.error("", e);
		}
	}

	protected String getPath(File file) {
		String path = file.getPath().replaceAll("\\\\", "/");
		return path.startsWith("/") ? path : "/" + path;
	}

	@Override
	public List<File> listAll(File file) {
		return list(a -> true, file);
	}

}
